package com.源码分析.processor;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.TopicProcessor;

/**TopicProcessor是一个异步的processor，当shared设置为true的时候，支持对多个publisher的并发重放。如果订阅的publisher是一个并发的stream或者是需要并发调用Topicrocessor的onNext,onCompleete,onError方法，则必须强制开启share。关闭share则是遵循reactive streams规范的processor，不允许并发调用。
TopicProcessor也支持把消息广播(fan-out)到多个subscriber，它给每个subscriber绑定一个线程。能够支持的subscriber的最大个数由线程池executor限制。
TopicProcessor使用了RingBuffer数据结构来推送数据，每个subscriber线程都在RingBuffer记录其消费的位置
TopicProcessor也支持autoCancel选项，默认为true，也就是当所有subscriber都取消订阅的时候，publisher也会被自动cannel

注意两个地方:
share
share背后设置的是EventLoopProcessor的multiproducers属性

如果share为true,则创建的是createMultiProducer.
具体的表象就是如果有多线程调用processor的onNext方法,而没有开启share的话,会有并发问题,即数据会丢失.比如上面的代码,如果注释掉share(true),则最后count的大小就不一定是100,而开启share为true就能保证最后count的大小是100
如果设置executor(Executors.newSingleThreadExecutor()),则flux1,flux2,flux3的订阅者则是顺序执行,而不是并发的.
 * @author zhangsulei
 *
 */
public class TopicProcessorDemo {

	static final Logger LOGGER = LoggerFactory.getLogger(TopicProcessorDemo.class);

	@Test
    public void testWorkQueueProcessor() throws InterruptedException{
		TopicProcessor<Integer> topicProcessor = TopicProcessor.<Integer>builder()
                .share(true)
//                .executor(Executors.newSingleThreadExecutor())
                .build();

        AtomicInteger count = new AtomicInteger(0);
        topicProcessor.subscribe(e -> {
            LOGGER.info("flux1 subscriber:{}",e);
            count.incrementAndGet();
        });
        topicProcessor.subscribe(e -> {
            LOGGER.info("flux2 subscriber:{}",e);
            count.incrementAndGet();
        });

        IntStream.rangeClosed(1,1000)
        .parallel()
        .forEach(e -> {
            topicProcessor.onNext(e);
        });

        topicProcessor.onComplete();
        topicProcessor.blockLast();

        TimeUnit.SECONDS.sleep(1);
        System.out.println(count.get());
	}

}
